package com.zhengb.flink.study.datastream.datasource;

import com.alibaba.fastjson.JSONObject;
import com.zhengb.flink.study.datastream.entity.GoodSell;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 模拟商品出售数据源
 *
 * @author zhengbo
 * @date 2019/12/20
 */
@Slf4j
public class GoodsSellDataSource extends RichParallelSourceFunction<GoodSell> {

    private static volatile boolean isRunning = true;

    private final static List<Double> unitPrice = Arrays.asList(19.99, 29.99, 39.98, 49.0, 87.0, 99.99);

    @Override
    public void run(SourceContext<GoodSell> ctx) throws Exception {

        while (isRunning) {

            ThreadLocalRandom random = ThreadLocalRandom.current();
            int type = random.nextInt(5);

            String goodName = "商品:" + (char) ('A' + type);
            int sellNum = random.nextInt(20);
            double totalAmount = sellNum * unitPrice.get(type);

            GoodSell sellRecord = new GoodSell(goodName, sellNum, totalAmount);

            log.info("good sell emit:{}", JSONObject.toJSONString(sellRecord));

            ctx.collectWithTimestamp(sellRecord, System.nanoTime());

            Thread.sleep(2000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
